package com.gagakuai.bpm.mqlistener.camunda;


import com.gagakuai.bpm.common.component.PublisherContext;
import com.gagakuai.bpm.common.constants.RabbitMqConstants;
import com.gagakuai.bpm.domian.events.DeadMessageEvent;
import com.gagakuai.bpm.domian.wapper.DeadMessageWrapper;
import com.gagakuai.bpm.mqlistener.camunda.baseproxy.BaseConsumer;
import com.gagakuai.bpm.mqlistener.camunda.baseproxy.BaseConsumerProxy;
import com.gagakuai.bpm.mqlistener.camunda.handler.returns.BpmReturnFirstStepConsumer;
import com.gagakuai.bpm.mqlistener.camunda.handler.returns.BpmReturnLastStepConsumer;
import com.gagakuai.bpm.mqlistener.camunda.handler.returns.BpmWithReturnStepConsumer;
import com.gagakuai.bpm.mqlistener.camunda.handler.sync.BpmSyncAccountConsumer;
import com.gagakuai.bpm.mqlistener.camunda.handler.withdraw.BpmWithDrawConsumer;
import com.gagakuai.bpm.utils.DateUtils;
import com.gagakuai.bpm.utils.RedisUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;
import java.util.Queue;

/*
 * @program: mss_server
 * @description: bpm的业务消息接收
 * @author: houhong
 * @create: 2022-03-24 15:09
 */
@Component
@Slf4j
public class BpmMessageBusnessListener implements BeanFactoryAware {


    @Autowired
    private RedisUtil msgLogApi;

    @Autowired
    private BpmWithDrawConsumer withDrawConsumer;

    @Autowired
    private BpmReturnLastStepConsumer bpmReturnLastStepConsumer;

    @Autowired
    private BpmReturnFirstStepConsumer bpmReturnFirstStepConsumer;

    @Autowired
    private BpmWithReturnStepConsumer bpmWithReturnStepConsumer;

    @Autowired
    private BpmSyncAccountConsumer bpmSyncAccountConsumer;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private PublisherContext publisherContext;

    @Autowired
    private Queue<DeadMessageWrapper> deadMessageEventQueue;

    private ConfigurableListableBeanFactory beanFactory;


    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {

        this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
    }


    /*
     * @param
     * @return
     * @Author houhong
     * @Description //TODO bpm消费业务 -- 撤销流程
     * @Date 3:20 下午 2022/3/24
     */
    @RabbitListener(queues = RabbitMqConstants.BPM_WITH_DRAW_QUEUE_NAME)
    public void consumerWithDrawmeaasge(Message message, Channel channel) throws Exception {


        log.info("#【bpm撤销队列】消费消息,时间:{}#", DateUtils.getCurrentTime());

        BaseConsumerProxy baseConsumerProxy = new BaseConsumerProxy(withDrawConsumer, msgLogApi);
        BaseConsumer proxy = (BaseConsumer) baseConsumerProxy.getProxy();
        if (Objects.nonNull(proxy)) {
            proxy.consume(message, channel, RabbitMqConstants.BPM_WITH_DRAW_QUEUE_NAME);
        }
    }

    /*
     * @param
     * @return
     * @Author houhong
     * @Description //TODO bpm消费业务 -- 取回流程、
     * @Date 3:20 下午 2022/3/24
     */
    @RabbitListener(queues = RabbitMqConstants.BPM_WITH_RETURN_QUEUE_NAME)
    public void consumerWithReturnMeaasge(Message message, Channel channel) throws Exception {


        log.info("#【bpm取回队列】消费消息,时间:{}#", DateUtils.getCurrentTime());

        BaseConsumerProxy baseConsumerProxy = new BaseConsumerProxy(bpmWithReturnStepConsumer, msgLogApi);
        BaseConsumer proxy = (BaseConsumer) baseConsumerProxy.getProxy();
        if (Objects.nonNull(proxy)) {
            proxy.consume(message, channel, RabbitMqConstants.BPM_WITH_RETURN_QUEUE_NAME);
        }

    }


    /*
     * @param
     * @return
     * @Author houhong
     * @Description //TODO bpm消费业务 -- 退回到上一步
     * @Date 3:20 下午 2022/3/24
     */
    @RabbitListener(queues = RabbitMqConstants.BPM_WITH_DRAW_LAST_STEP_QUEUE_NAME)
    public void consumerWithDrawLastStepmeaasge(Message message, Channel channel) throws Exception {


        log.info("#【bpm退回到上一步队列】消费消息,时间:{}#", DateUtils.getCurrentTime());
        BaseConsumerProxy baseConsumerProxy = new BaseConsumerProxy(bpmReturnLastStepConsumer, msgLogApi);
        BaseConsumer proxy = (BaseConsumer) baseConsumerProxy.getProxy();
        if (Objects.nonNull(proxy)) {
            proxy.consume(message, channel, RabbitMqConstants.BPM_WITH_DRAW_LAST_STEP_QUEUE_NAME);
        }

    }


    /*
     * @param
     * @return
     * @Author houhong
     * @Description //TODO bpm消费业务 -- 退回到第一步
     * @Date 10:57 上午 2022/4/06
     */
    @RabbitListener(queues = RabbitMqConstants.BPM_WITH_DRAW_FIRST_STEP_QUEUE_NAME)
    public void consumerWithDrawFirstStepmeaasge(Message message, Channel channel) throws Exception {


        log.info("#【bpm退回到第一步队列】消费消息,时间:{}#", DateUtils.getCurrentTime());


      /*  MessageProperties properties = message.getMessageProperties();
        long tag = properties.getDeliveryTag();
        channel.basicAck(tag,false);*/
        BaseConsumerProxy baseConsumerProxy = new BaseConsumerProxy(bpmReturnFirstStepConsumer, msgLogApi);
        BaseConsumer proxy = (BaseConsumer) baseConsumerProxy.getProxy();
        if (Objects.nonNull(proxy)) {
            proxy.consume(message, channel, RabbitMqConstants.BPM_WITH_DRAW_FIRST_STEP_QUEUE_NAME);
        }

    }

    /*
     * @param
     * @return
     * @Author houhong
     * @Description //TODO bpm 同步人员信息
     * @Date 10:57 上午 2022/4/06
     */
    @RabbitListener(queues = RabbitMqConstants.BPM_WITH_SYNC_ACCOUNT_QUEUE_NAME)
    public void consumerSyncAccountStepmeaasge(Message message, Channel channel) throws Exception {


        log.info("#【bpm同步人员信息队列】消费消息,时间:{}#", DateUtils.getCurrentTime());

        bpmSyncAccountConsumer.consume(message, channel, RabbitMqConstants.BPM_WITH_SYNC_ACCOUNT_QUEUE_NAME);
    }

    /*
     * @param message
     * @param channel
     * @return
     * @Author houhong
     * @Description //TODO 死信队列消费
     * @Date 2:24 下午 2022/7/8
     */
    @RabbitListener(queues = RabbitMqConstants.BPM_DEAD_QUEUE_NAME)
    public void consumerdeadMeaasge(Message message, Channel channel) throws Exception {

        //使用异步线程
        log.info("死信队列收到消息 :{} ", message.toString());


        //发布事件出去
        DeadMessageEvent.DeadMessageEventData deadMessageEventData = new DeadMessageEvent.DeadMessageEventData();
        deadMessageEventData.setMessage(message);
        deadMessageEventData.setChannel(channel);
        deadMessageEventData.setCorrelationId(getCorrelationId(message));
        DeadMessageEvent deadMessageEvent = new DeadMessageEvent(deadMessageEventData);

        publisherContext.publish(deadMessageEvent);

    }

    /*
     * 获取CorrelationId
     *
     * @param message
     * @return
     */
    private String getCorrelationId(Message message) {

        String correlationId = null;

        MessageProperties properties = message.getMessageProperties();
        Map<String, Object> headers = properties.getHeaders();
        for (Map.Entry entry : headers.entrySet()) {
            String key = (String) entry.getKey();
            if ("spring_returned_message_correlation".equals(key)) {
                String value = (String) entry.getValue();
                correlationId = value;
                return correlationId;
            }
        }

        return null;
    }


}